/*
 * LocalDispatchingLayerProxy.java   0.0.1    Nov 17, 2009
 * 
 * Copyright (c) 2009 mentalsmash.org
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * The use of the Apache License does not indicate that this project is
 * affiliated with the Apache Software Foundation.
 */
package org.mentalsmash.tazio.net.ril.impl;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.mentalsmash.tazio.net.ril.DispatchingLayerProxy;
import org.mentalsmash.tazio.net.ril.RemoteInvoker;
import org.mentalsmash.tazio.net.ril.exceptions.NoResponseReceivedException;
import org.mentalsmash.tazio.net.ril.messages.MethodInvocationRequestMessage;
import org.mentalsmash.tazio.net.ril.messages.MethodInvocationResponseMessage;
import org.mentalsmash.tazio.net.ril.messages.RILRequestMessage;
import org.mentalsmash.tazio.net.ril.messages.RILResponseMessage;
import org.mentalsmash.tazio.net.ril.messages.RegistryContentRequestMessage;
import org.mentalsmash.tazio.net.ril.messages.RemoteObjectRequestMessage;
import org.mentalsmash.tazio.commons.TazioRuntimeException;
import org.mentalsmash.tazio.commons.identifiers.OID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.inject.Inject;
import com.google.inject.name.Named;

/**
 * @version 0.0.1 Nov 17, 2009
 * @author Andrea Sorbini <as@mentalsmash.org>
 * 
 */
public class LocalDispatchingLayerProxy implements DispatchingLayerProxy {

    private final static Logger log = LoggerFactory
	    .getLogger(LocalDispatchingLayerProxy.class);

    private RemoteInvoker localhost = null;
    private boolean inGroup = false;
    
    private final OID oid = OID.generateOID();
    
    
    private final Map<OID, Set<RemoteInvoker>> objsToInvoker = new HashMap<OID, Set<RemoteInvoker>>();
    private final Map<RemoteInvoker, Integer> objsCounts = new HashMap<RemoteInvoker, Integer>();

    // private final Set<RemoteInvoker> group = new HashSet<RemoteInvoker>();
    // private final Set<RemoteInvoker> remoteHosts = new
    // HashSet<RemoteInvoker>();

    /* (non-Javadoc)
     * @see java.lang.Object#hashCode()
     */
    @Override
    public int hashCode() {
	final int prime = 31;
	int result = 1;
	result = prime * result + ((oid == null) ? 0 : oid.hashCode());
	return result;
    }

    /* (non-Javadoc)
     * @see java.lang.Object#equals(java.lang.Object)
     */
    @Override
    public boolean equals(Object obj) {
	if (this == obj)
	    return true;
	if (obj == null)
	    return false;
	if (getClass() != obj.getClass())
	    return false;
	LocalDispatchingLayerProxy other = (LocalDispatchingLayerProxy) obj;
	if (oid == null) {
	    if (other.oid != null)
		return false;
	} else if (!oid.equals(other.oid))
	    return false;
	return true;
    }

    private final Set<LocalDispatchingLayerProxy> group = new HashSet<LocalDispatchingLayerProxy>();
    private final Set<LocalDispatchingLayerProxy> remoteHosts = new HashSet<LocalDispatchingLayerProxy>();

    @Inject
    public LocalDispatchingLayerProxy(RemoteInvoker localhost,
	    @Named("inGroup") boolean inGroup) {
	this.localhost = localhost;

	this.inGroup = inGroup;
	
	if (inGroup) this.group.add(this);

	// this.localhost.setDispatchingLayerProxy(this);
    }

    public void setLocalhost(RemoteInvoker localhost) {
	this.localhost = localhost;
    }
    
    public void connectTo(LocalDispatchingLayerProxy proxy) {

	if (!this.remoteHosts.contains(proxy)) this.remoteHosts.add(proxy);

	if (proxy.inGroup && !this.group.contains(proxy)) {
	    this.group.add(proxy);
	}

	if (!proxy.remoteHosts.contains(this)) {
	    proxy.connectTo(this);
	}
    }

    public void disconnectFrom(LocalDispatchingLayerProxy proxy) {

	this.remoteHosts.remove(proxy);
	this.group.remove(proxy);

	if (proxy.remoteHosts.contains(this)) {
	    proxy.disconnectFrom(this);
	}

    }
    
    private void addObjectsMapping(RemoteInvoker invoker) {
	
	OID[] objs = invoker.getRegistry().getRegisteredOIDs();
	
	for (OID oid : objs) {
	    
	    Set<RemoteInvoker> invokers = this.objsToInvoker.get(oid);
	    
	    if (invokers == null){
		invokers = new HashSet<RemoteInvoker>();
		this.objsToInvoker.put(oid, invokers);
	    }
	    
	    invokers.add(invoker);
	    
	    log.debug("Registered "+oid+" on "+invoker);
	}
	
    }
    
    private void checkObjsCounts(RemoteInvoker invoker) {
//	log.debug("Checking objects on "+invoker);
	Integer counts = this.objsCounts.get(invoker);
	
	if (counts == null) {
	    counts = 0;
	    this.objsCounts.put(invoker, counts);
	}
	
	if (counts > 0) {
	    return;
	}
	
	int objsNum = invoker.getRegistry().getRegisteredOIDs().length;
	
	log.debug("There are "+objsNum+" objects on invoker "+invoker+" (registered: "+counts+")");
	
	if (objsNum > counts) {
	    this.addObjectsMapping(invoker);
	    this.objsCounts.put(invoker, objsNum);
	}
	
    }
    
    private void checkRemoteHosts(){
	
//	if (this.objsCounts.values().iterator().next() > 0) return;
//	
	
	for (LocalDispatchingLayerProxy remoteHost : this.remoteHosts) {
	    RemoteInvoker invoker = remoteHost.localhost;
	    this.checkObjsCounts(invoker);
	}
    }
    
    @SuppressWarnings("unchecked")
    private <T extends Serializable> T simulateNetwork(T obj) {
	try {
	    ByteArrayOutputStream bout = new ByteArrayOutputStream();
	    ObjectOutputStream oout = new ObjectOutputStream(bout);
	    oout.writeObject(obj);
	    ByteArrayInputStream bin = new ByteArrayInputStream(bout
		    .toByteArray());
	    ObjectInputStream oin = new ObjectInputStream(bin);
	    obj = (T) oin.readObject();
	    return obj;
	} catch (Exception e) {
	    e.printStackTrace();
	    throw new TazioRuntimeException(e);
	}
    }

    @Override
    public RILResponseMessage sendMessage(RILRequestMessage request)
	    throws NoResponseReceivedException {

	request = this.simulateNetwork(request);

	RemoteInvoker recipient = this.findRecipient(request);

	if (recipient == null) {
	    throw new NoResponseReceivedException(request);
	}

	log.trace("Delivering message " + request + " to " + recipient);
	RILResponseMessage response = recipient.getMessageHandler()
		.handleRequest(request);
	log.trace("Got response : " + response);

	response = this.simulateNetwork(response);

	return response;
    }

    @Override
    public MethodInvocationResponseMessage sendMethodInvocationAtMostOnce(
	    MethodInvocationRequestMessage request)
	    throws NoResponseReceivedException {

	return (MethodInvocationResponseMessage) this.sendMessage(request);
    }

    @Override
    public MethodInvocationResponseMessage sendMethodInvocationToGroup(
	    MethodInvocationRequestMessage request)
	    throws NoResponseReceivedException {
	
	request = this.simulateNetwork(request);

	MethodInvocationResponseMessage response = null;
	
	for (LocalDispatchingLayerProxy remoteHost : this.group) {
	    RemoteInvoker invoker = remoteHost.localhost;
	    MethodInvocationResponseMessage resp = (MethodInvocationResponseMessage) invoker
		    .getMessageHandler().handleRequest(request);

	    if (response == null) {
		response = resp;
	    }

	}

	response = this.simulateNetwork(response);
	
	return response;
    }

    private RemoteInvoker findRecipient(RILRequestMessage request) {
	
	if (request instanceof MethodInvocationRequestMessage) {
	    return findRecipient((MethodInvocationRequestMessage) request);
	} else if (request instanceof RemoteObjectRequestMessage) {
	    return findRecipient((RemoteObjectRequestMessage) request);
	} else if (request instanceof RegistryContentRequestMessage) {
	    return findRecipient((RegistryContentRequestMessage) request);
	}

	return null;
    }

    /**
     * @param msg
     */
    private RemoteInvoker findRecipient(MethodInvocationRequestMessage msg) {
	this.checkRemoteHosts();
	
	OID objectOID = msg.getRemoteObject();

//	for (LocalDispatchingLayerProxy remoteHost : this.remoteHosts) {
//	    RemoteInvoker invoker = remoteHost.localhost;
//	    if (invoker.getRegistry().hasObject(objectOID)) {
//
//		return invoker;
//	    }
//	}
//
//	return null;
	
	Set<RemoteInvoker> invokers = this.objsToInvoker.get(objectOID);
	
	if (invokers != null)
	    return invokers.iterator().next();
	else
	    return null;
	
    }

    /**
     * @param msg
     */
    private RemoteInvoker findRecipient(RegistryContentRequestMessage msg) {

	for (LocalDispatchingLayerProxy remoteHost : this.remoteHosts) {
	    RemoteInvoker invoker = remoteHost.localhost;
	    if (invoker.getOID().equals(msg.getRemoteInvokerOID())) {

		return invoker;

	    }
	}

	return null;

    }

    /**
     * @param msg
     */
    private RemoteInvoker findRecipient(RemoteObjectRequestMessage msg) {
	OID objectOID = msg.getRequested();

	for (LocalDispatchingLayerProxy remoteHost : this.remoteHosts) {
	    RemoteInvoker invoker = remoteHost.localhost;
	    if (invoker.getRegistry().hasObject(objectOID)) {

		return invoker;

	    }
	}

	return null;

    }

}
